package com.tarena.emqx.demo.consumer;

import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;

/**
 * 在spring容器中 使用这个类的对象
 * 实现订阅主题监听消息的消费功能
 */
@Slf4j
public class EmqxConsumer implements MqttCallbackExtended {
    //在消费对象中 利用不同的调用方法的时间点 使用mqttClient和server端进行通信
    private MqttClient mqttClient;
    public EmqxConsumer(MqttClient mqttClient){
        this.mqttClient=mqttClient;
    }
    //1. 连接成功的时候 connectComplete
    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        //可以使用 客户端client 正常监听一个主题
        try {
            mqttClient.subscribe("/emqx/first/topic");
        } catch (MqttException e) {
            log.info("监听订阅主题失败",e);
        }
    }
    //2. 连接建立成功后链接断开 connectionLost
    @Override
    public void connectionLost(Throwable throwable) {
        //订阅链接一旦断开
        log.error("监听订阅主题链接断开",throwable);
    }
    //3. 连接建立成功后消息投递达到 messageArrived 核心消费逻辑
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        //1.topic: 消息到达 告知这个方法 消息来源 是哪个主题
        //2.message: 投递过来的消息对象
        //测试案例的消费逻辑 打印 生产者发送过来的消息数据
        byte[] body = message.getPayload();//载荷 就是发送消息体byte[]数组
        String msg=new String(body, StandardCharsets.UTF_8);
        log.info("消费者消费消息:{},订阅主题:{}",msg,topic);
    }
    //4. 连接建立成功后消息消费结束 deliveryComplete
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        log.info("消费者消费消息 投递结束");
    }
}
